package bsf.baseservice.businessmq.consumer;


import bsf.base.CallBack;
import bsf.baseservice.businessmq.common.BusinessMQConfig;
import bsf.baseservice.businessmq.db.*;
import bsf.baseservice.businessmq.common.BusinessMQResponse;
import bsf.baseservice.businessmq.model.*;
import bsf.baseservice.businessmq.systemruntime.*;
import bsf.baseservice.businessmq.systemruntime.log.*;

import bsf.db.DbConn;
import bsf.util.ArrayUtil;

import java.util.ArrayList;
import java.util.Collections;

/**
 消费者使用提供类
 
*/
public class ConsumerProvider implements AutoCloseable
{
	public BusinessMQConfig Config;
	public final String getMQPath()
	{
		return mqpath;
	}
	public final void setMQPath(String value)
	{
		mqpath = value.toLowerCase().trim();
	}
	protected String mqpath;
	/** 
	 当前客户端业务的唯一标识
	 
	*/
	public final String getClient()
	{
		return client;
	}
	public final void setClient(String value)
	{
		client = value.toLowerCase().trim();
	}
	protected String client;
	/** 
	 客户端名称,仅用于显示区分
	 
	*/
	public String ClientName = "未定义";
	/** 
	 消费者处理线程数量（默认自动设置为当前可用分区数量，为性能最佳，过多或者过少，将被系统重置,调试时可以设置为1）
	 
	*/
	public int MaxReceiveMQThread = 0;
	public ArrayList<Integer> PartitionIndexs = new ArrayList<Integer>();

	private ConsumerContext Context;

	public ConsumerProvider()
	{

	}
	/** 
	 注册消息循环
	 
	*/
	public final <T> void registerReceiveMQListener(final CallBack.Action1<BusinessMQResponse<T>> action, final Class<T> cls)
	{
		try
		{
			if (Context != null)
			{
				throw new BusinessMQException("当前实例不能打开多个Consumer监听");
			}

			PartitionIndexs = (ArrayList<Integer>) ArrayUtil.distinct(PartitionIndexs);
			PartitionIndexs =  (ArrayList<Integer>) ArrayUtil.sort(PartitionIndexs);
			//PartitionIndexs =  (from o in PartitionIndexs orderby o select o).Distinct().ToList();
			Context = new ConsumerContext();
			Context.ConsumerProvider=this;
			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterReceiveMQListener", "消费者开始注册消息回调");
			//注册信息
			registerConsumerInfo();
			//注册消息回调
			Context.Listener=new ReceiveMessageListener();
			ConsumerActionInfo t = new ConsumerActionInfo();
			t.Action=action;
			t.ReturnType=cls;
			Context.ActionInfo=t;
			Context.Listener.register(new CallBack.Func1<Object,MQMessage>() {
				@Override
				public Object invoke(MQMessage message) {
					BusinessMQResponse<T> r = new BusinessMQResponse<T>();
					r.InnerObject=message;
					r.ObjMsg=message.<T>getMessageObj(cls);
					action.invoke(r);
					return null;
				}
			}
		   , Context);
			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterReceiveMQListener", "注册消费者监听成功");
			LogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterReceiveMQListener", "注册消费者监听成功");

		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterReceiveMQListener", "消费者注册MQ监听出错", exp);

			try
			{
				this.close();
			}
			catch (Exception e)
			{
			}
			throw exp;
		}
	}

	private void registerConsumerInfo()
	{
		ConsumerBLL bll = new ConsumerBLL();
		DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterConsumerInfo", "消费者开始注册消费者信息");
		//注册消费者
				//c.BeginTransaction();
				//取消注册用户信息
				//注册并获取clientid
				//清理过期不心跳消费端,并检验partitionindex不重复,并注册消费者
				//消费者订阅队列信息
				//注册并更新消费者端分区信息
				//获取分区节点信息缓存
				//校准服务器时间
				//c.Commit();
				//c.Rollback();
		try(DbConn c =DbConn.createConn(Config.ManageConnectString))
		{
			try
			{
				if (Context.ConsumerInfo != null && Context.ConsumerInfo.ConsumerModel != null)
				{
					bll.removeConsumer(c, Context.ConsumerInfo.ConsumerModel.tempid, Context.ConsumerInfo.ConsumerModel.consumerclientid);
				}
				Context.ConsumerInfo=new ConsumerInfo();
				Context.ConsumerInfo.ConsumerPartitionModels=new ArrayList<>();
				Context.ConsumerInfo.ConsumerClientModel=bll.registerClient(c, getClient());
				if (Context.ConsumerInfo.ConsumerClientModel == null)
				{
					throw new BusinessMQException("客户端注册client失败！");
				}
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterConsumerInfo", "注册并获取clientid成功");
				Context.ConsumerInfo.ConsumerModel=bll.registerConsumer(c, Context.ConsumerInfo.ConsumerClientModel.id, ClientName, PartitionIndexs);
				if (Context.ConsumerInfo.ConsumerModel == null)
				{
					throw new BusinessMQException("当前客户端注册consumer失败！");
				}
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterConsumerInfo", "清理过期不心跳消费端,并检验partitionindex不重复,并注册消费者成功");
				Context.ConsumerInfo.MQPathModel=bll.getMQPath(c, this.getMQPath());
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterConsumerInfo", "消费者订阅队列信息");
				for (Integer partitionindex : PartitionIndexs)
				{
					tb_consumer_partition_model model = bll.registerConsumerPartition(c, Context.ConsumerInfo.ConsumerClientModel.id, partitionindex, getMQPath(), Context.ConsumerInfo.ConsumerModel.tempid);
					if (model != null)
					{
						Context.ConsumerInfo.ConsumerPartitionModels.add(model);
					}
				}
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterConsumerInfo", "注册并更新消费者端分区信息");
				ArrayList<Integer> datanodepartition = new ArrayList<Integer>();
				for (tb_consumer_partition_model p : Context.ConsumerInfo.ConsumerPartitionModels)
				{
					PartitionIDInfo partitionidinfo = PartitionRuleHelper.getPartitionIDInfo(p.partitionid);
					if (!datanodepartition.contains(partitionidinfo.DataNodePartition))
					{
						datanodepartition.add(partitionidinfo.DataNodePartition);
					}
				}
				Context.ConsumerInfo.DataNodeModelDic=bll.getDataNodeModelsDic(c, datanodepartition);
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterConsumerInfo", "获取分区节点信息缓存");
				Context.setManageServerTime(c.getServerDate());
			}
			catch (Exception exp)
			{
				throw exp;
			}
		}

		DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterConsumerInfo", "注册消费者信息完毕");
		//获取系统配置信息
		ConfigHelper.loadConfig(Config.ManageConnectString);
		DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "RegisterConsumerInfo", "获取系统配置信息");
	}


	public final void close()
	{
		if (Context != null)
		{
			try
			{
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "Dispose", "开始消费者资源释放");
				if (Context != null)
				{
					Context.close();
				}
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "Dispose", "消费者上下文资源释放成功");
			}
			catch (Exception exp)
			{
				ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "Dispose", "消费者资源释放出错", exp);
				throw new BusinessMQException("释放订阅客户端消息处理资源失败", exp);
			}
			finally
			{
				try
				{
					ConsumerBLL bll = new ConsumerBLL();
					//取消注册
					try(DbConn c = DbConn.createConn(Config.ManageConnectString))
					{
						bll.removeConsumer(c, Context.ConsumerInfo.ConsumerModel.tempid, Context.ConsumerInfo.ConsumerModel.consumerclientid);
					}
					DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "Dispose", "消费者注销注册信息成功");
					LogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "Dispose", "消费者资源释放成功");
				}
				catch (Exception exp1)
				{
					ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "Dispose", "消费者资源释放出错", exp1);
					throw new BusinessMQException("移除订阅客户端注册信息失败", exp1);
				}
				finally
				{
					Context = null;
				}
			}

		}
	}
}